package com.lagou.observer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;

public class MyObserver extends BaseRegionObserver {
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        // 获取table对象
//        HTableInterface friends = e.getEnvironment().getTable(TableName.valueOf("friends"));
        Configuration configuration = e.getEnvironment().getConfiguration();
        Connection conn = ConnectionFactory.createConnection(configuration);
        Table friends = conn.getTable(TableName.valueOf("friends"));
        // 获取要删除的rowkey
        byte[] row = delete.getRow();
        // 获取要删除的所有cell对象
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        Set<Map.Entry<byte[], List<Cell>>> entries = familyCellMap.entrySet();
        // 遍历所有的cell，删除相应的信息
        for(Map.Entry<byte[], List<Cell>> entry: entries){
            // 获取要删除的所有列
            List<Cell> cells = entry.getValue();
            for(Cell cell: cells){
                // 获取rowkey信息
                byte[] rowkey = CellUtil.cloneRow(cell);
                // 获取列信息
                byte[] column = CellUtil.cloneQualifier(cell);
                boolean flag = friends.exists(new Get(column).addColumn(Bytes.toBytes("friends"), rowkey));
                if(flag){
                    System.out.println("deleting" + column + ": " + rowkey);
                    Delete co_delete = new Delete(column).addColumn(Bytes.toBytes("friends"), rowkey);
                    friends.delete(co_delete);
                }else{
                    System.out.println(column + ": " + rowkey + " is already deleted");
                }
            }
        }
        friends.close();
    }
}
